package com.example.client;

import com.example.decoder.RpcDecoder;
import com.example.encoder.RpcEncoder;
import com.example.handler.UserServiceClientHandler;
import com.example.quartz.QuartzMain;
import com.example.request.RpcRequest;
import com.example.serialization.JSONSerializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import org.I0Itec.zkclient.ZkClient;

import java.lang.reflect.Proxy;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.List;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author lbq
 * @version 1.0
 * @date 2021/2/4 21:28
 */
public class RPCConsumer {

    private static final ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    private static UserServiceClientHandler userServiceClientHandler;

    private final static ZkClient zkClient = new ZkClient("127.0.0.1:2181");

    private Integer round = 1;

    private static String[] server;

    public void initClient() throws Exception {
        userServiceClientHandler = new UserServiceClientHandler();

        NioEventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(new StringDecoder());
                        pipeline.addLast(userServiceClientHandler);
                    }
                });
        List<String> children = zkClient.getChildren("/zkServer");
        server = children.toArray(new String[0]);
        // 随机负载均衡策略
        Random random = new Random();
        int index = random.nextInt(2);
        System.out.println("index = " + index);
        // 轮询负载均衡策略
        // int index = round % server.length;
        // System.out.println("index = " + index);
        String s = server[index];
        String connect = zkClient.readData("/zkServer/" + s);
        System.out.println("connect = " + connect);
        String[] split = connect.split(":");
        System.out.println("客户端初始化,连接服务端地址为: " + s);
        bootstrap.connect(split[0], Integer.parseInt(split[1])).sync();
        round++;
    }

    public Object createProxy(Class<?> serviceClass) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class[]{serviceClass}, (proxy, method, args) -> {
                    initClient();
                    RpcRequest rpcRequest = new RpcRequest();
                    String uuid = UUID.randomUUID().toString();
                    System.out.println(uuid);
                    rpcRequest.setRequestId(uuid);
                    String className = serviceClass.getName();
                    System.out.println(className);
                    rpcRequest.setClassName(className);
                    rpcRequest.setMethodName(method.getName());
                    Class<?>[] parameterTypes = method.getParameterTypes();
                    rpcRequest.setParameterTypes(parameterTypes);
                    rpcRequest.setParameters(args);
                    userServiceClientHandler.setParam(rpcRequest);
                    return executorService.submit(userServiceClientHandler).get();
                });
    }

    public void loadServer() throws Exception {
        List<String> serverList = zkClient.getChildren("/zkServer");
        // System.out.println("serverList.size() = " + serverList.size());
        // System.out.println("server.length = " + server.length);
        if (serverList.size() < server.length) {
            server = serverList.toArray(new String[0]);
            System.out.println("服务端节点数量减少");
            initClient();
            notifyAll();
        } else if (serverList.size() > server.length) {
            server = serverList.toArray(new String[0]);
            System.out.println("服务端节点数量增加");
            initClient();
        }
        List<String> children = zkClient.getChildren("/zkServerHeartbeat");
        System.out.println("定时更新节点信息");
        for (String child : children) {
            // System.out.println(child);
            Long time = zkClient.readData("/zkServerHeartbeat/" + child);
            // System.out.println("time = " + time);
            long now = LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli();
            // System.out.println("now = " + now);
            // System.out.println(now - time > 5000);
            if (now - time > 5000 && zkClient.exists("/zkServer/" + child)) {
                System.out.println("删除节点 : " + "/zkServer/" + child);
                zkClient.delete("/zkServer/" + child);
                try {
                    initClient();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
